package com.test.study.ToolsUtils.flume;

import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;

/**
 * 自定一个sink
 * 
 * @author huyong
 *
 */
public class MySink extends AbstractSink implements Configurable {

	public void configure(Context context) {
		System.out.println("读取配置");
		System.out.println(context.getString("wanghaha"));

	}

	/**
	 * 核心的处理方法
	 */
	public Status process() throws EventDeliveryException {
		Status status = null;

		// Start transaction
		Channel ch = getChannel();
		Transaction txn = ch.getTransaction();
		txn.begin();
		try {
			// This try clause includes whatever Channel operations you want to do

			Event event = ch.take();
			Map<String, String> headers = event.getHeaders();
			Set<Entry<String, String>> entrySet = headers.entrySet();
			Iterator<Entry<String, String>> iterator = entrySet.iterator();

			while (iterator.hasNext()) {
				Entry<String, String> next = iterator.next();
				System.out.println(next.getKey() + "," + next.getValue());
			}

			// Send the Event to the external repository.
			// storeSomeData(e);

			txn.commit();
			status = Status.READY;
		} catch (Throwable t) {
			txn.rollback();

			// Log exception, handle individual exceptions as needed

			status = Status.BACKOFF;

			// re-throw all Errors
			if (t instanceof Error) {
				throw (Error) t;
			}
		}
		return status;
	}

	/**
	 * 初始化连接 只执行一次
	 */
	@Override
	public synchronized void start() {
		// TODO Auto-generated method stub
		super.start();
	}

	/**
	 * 关闭连接 只执行一次
	 */
	@Override
	public synchronized void stop() {
		// TODO Auto-generated method stub
		super.stop();
	}

}
